/*
 * Copyright 2010 Thedwick, LLC

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.

 */
package com.thedwick.jdbcGrabber.sink;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import com.thedwick.jdbcGrabber.util.ClassUtil;
import com.thedwick.jdbcGrabber.util.ThreadUtil;

/**
 * Efficiently keeps track of SQL statements that have been executed and also the stack 
 * that executed them.
 * 
 * Only keeps a hash of the SQL statements in memory, not the entire SQL statement.  Writes out
 * a table of SQL statements periodically to a file, as designated in the properties file.  
 *
 */
public class SummarizingSink implements SQLSink {
	private final SummaryStore store;
	private volatile Map<Integer, NormalizedSQL> sqlTable;
	private volatile Map<Integer, StackTraceElement[]> traceTable;
	private final Map<Integer, Long> sqlCounts;
	private final Map<Integer, Long> traceCounts;
	private long lastDumpTime = System.currentTimeMillis();
	private final long dumpInterval;
	private final Map<Integer, Set<Integer>> sqlToTrace;
	
	public SummarizingSink(Properties config){
		this(createStore(config), 30000);
	}
	
	public SummarizingSink(SummaryStore store, long dumpInterval){
		this.store = store;
		this.dumpInterval = dumpInterval;
		this.lastDumpTime = System.currentTimeMillis();
		sqlTable = new ConcurrentHashMap<Integer, NormalizedSQL>();
		traceTable = new ConcurrentHashMap<Integer, StackTraceElement[]>();
		sqlCounts = new ConcurrentHashMap<Integer, Long>();
		traceCounts = new ConcurrentHashMap<Integer, Long>();
		sqlToTrace = new ConcurrentHashMap<Integer, Set<Integer>>();
	}

	private static SummaryStore createStore(Properties config) {
		return (SummaryStore) ClassUtil.createInstance("summarizingSink.store", config, config);
	}
	
	@Override
	public void consume(String sql) {
		consume(new TraceElement(ThreadUtil.getStackTrace(), sql));
	}

	@Override
	public void consume(TraceElement trace){
		NormalizedSQL normalizedSql = consumeSql(trace.getSql());
		consumeStackTrace(normalizedSql, trace);
		dumpStatsIfNecessary();
	}

	private NormalizedSQL consumeSql(String sql) {
		NormalizedSQL normalizedSql = normalize(sql);
		countSQLPart(normalizedSql);
		return normalizedSql;
	}

	private void consumeStackTrace(NormalizedSQL normalizedSql, TraceElement trace) {
		Integer sqlKey = normalizedSql.getKey();
		Integer traceKey = trace.calculateTraceKey();
		synchronized(this){
			Long count = traceCounts.get(traceKey);
			if (count == null){
				traceTable.put(traceKey, trace.getStackTrace());
				count = new Long(1);
				traceCounts.put(traceKey, count);
			} else {
				traceCounts.put(traceKey, count.longValue() + 1);
			}
			Set<Integer> traceKeys = sqlToTrace.get(sqlKey);
			if (traceKeys == null){
				traceKeys = new HashSet<Integer>(2);
				sqlToTrace.put(sqlKey, traceKeys);
			}
			traceKeys.add(traceKey);
		}
	}

	private void countSQLPart(NormalizedSQL normalizedSql) {
		Integer key = normalizedSql.getKey();
		synchronized(this){
			Long count = sqlCounts.get(key);
			if (count == null){
				sqlTable.put(key, normalizedSql);
				count = new Long(1);
				sqlCounts.put(key, count);
			} else {
				sqlCounts.put(key, count.longValue() + 1);
			}
		}
	}

	private void dumpStatsIfNecessary() {
		if (System.currentTimeMillis() - lastDumpTime > dumpInterval){
			this.lastDumpTime = System.currentTimeMillis();
			dumpStats();
		}
	}

	
	/**
	 * Causes the accumulated counts and SQL table to be written out to the SQL store.
	 */
	void dumpStats() {
		try {
			Map<Integer, NormalizedSQL> sqlTable = null;
			Map<Integer, StackTraceElement[]> traceTable = null;
			synchronized(this){
				sqlTable = this.sqlTable;
				traceTable = this.traceTable;
				this.sqlTable = new ConcurrentHashMap<Integer, NormalizedSQL>();
				this.traceTable = new ConcurrentHashMap<Integer, StackTraceElement[]>();
				this.store.updateSQLCounts(this.sqlCounts);//not taking a copy of counts because, unlike the sql table, we want counts to be cumulative and not just the delta since the last dump
				this.store.updateTraceCounts(this.traceCounts);//not taking a copy of counts because, unlike the sql table, we want counts to be cumulative and not just the delta since the last dump
				this.store.storeTraceMapping(this.sqlToTrace);
			}
			this.store.storeSQLTable(sqlTable); //this can be outside the synchronized block because it's a copy
			this.store.storeTraceTable(traceTable); //this can be outside the synchronized block because it's a copy
		} catch (IOException e){
			e.printStackTrace();
		}
	}

	/**
	 * Contains the logic for 'normalizing' two strings that should be equivalent for 
	 * counting purposes.  For example, these two should be equivalent:
	 * select * from widgets where id = 1
	 * select * from widgets where id = 9999
	 * 
	 * so when fed the above statements, this method will return:
	 * select * from widgets where id = ?
	 * 
	 * Override this method if you want to just customize the logic for conflating statements
	 * together but otherwise want to do what is normally done.
	 * 
	 * @param sql
	 * @return
	 */
	protected NormalizedSQL normalize(String sql) {
		String rv = sql.replaceAll("-", "");//normalize negative numbers away
		rv = rv.replaceAll("\\b[0-9]+[\\.]*[0-9]*\\b", "?");//replace numbers with a single ?
		rv = rv.replaceAll("[\'\"][^\\s]*[\'\"]", "?");//replace strings with a single ?
		return new NormalizedSQL(rv); 
	}
}